package io.kiki.sba.registry.common.model.store;

import com.fasterxml.jackson.annotation.JsonIgnore;
import io.kiki.sba.registry.collections.Maps;
import io.kiki.sba.registry.common.model.ElementType;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.core.model.Scope;
import io.kiki.sba.registry.util.StringUtil;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

@Setter
@Getter
public class Subscriber extends BaseInfo {

    private static final long serialVersionUID = 1L;

    private Scope scope;

    private ElementType elementType;

    private String[] acceptEncodes;

    private boolean acceptMulti;

    /**
     * last push context
     */
    private Map<String /*dataCenter*/, PushContext> dataCenterToPushContextMap;

    private int pushedFailCount = 0;

    private long lastPushedFailTimeStamp = 0;

    /**
     * change subscriber word cache
     */
    public static Subscriber internSubscriber(Subscriber subscriber) {
        subscriber.setDataInfoId(subscriber.getDataInfoId());
        subscriber.setInstanceId(subscriber.getInstanceId());
        subscriber.setGroup(subscriber.getGroup());
        subscriber.setDataId(subscriber.getDataId());
        subscriber.setZone(subscriber.getZone());
        subscriber.setProcessId(subscriber.getProcessId());
        subscriber.setAppName(subscriber.getAppName());

        subscriber.setClientUrl(Url.internURL(subscriber.getClientUrl()));
        subscriber.setServerUrl(Url.internURL(subscriber.getServerUrl()));
        subscriber.setAttributes(subscriber.getAttributes());

        return subscriber;
    }


    private PushContext getPushContext(String dataCenter) {
        PushContext pushContext = null;
        if (dataCenterToPushContextMap == null) {
            pushContext = new PushContext();
            this.dataCenterToPushContextMap = Collections.singletonMap(dataCenter, pushContext);
        } else {
            pushContext = dataCenterToPushContextMap.get(dataCenter);
            if (pushContext == null) {
                // multi dataCenter, replace the singletonMap
                if (!(dataCenterToPushContextMap instanceof HashMap)) {
                    this.dataCenterToPushContextMap = new HashMap<>(dataCenterToPushContextMap);
                }
                pushContext = new PushContext();
                this.dataCenterToPushContextMap.put(dataCenter, pushContext);
            }
        }
        return pushContext;
    }


    /**
     * return true if one of any dataCenter need to update
     */
    public synchronized boolean checkVersion(@NonNull Map<String, Long> dataCenterToSubDatumVersionMap) {
        for (Entry<String, Long> entry : dataCenterToSubDatumVersionMap.entrySet()) {
            if (checkVersion(entry.getKey(), entry.getValue())) {
                return true;
            }
        }
        return false;
    }

    // check the version
    public synchronized boolean checkVersion(String dataCenter, long version) {
        final PushContext pushContext = getPushContext(dataCenter);
        /**
         * emptyVersion != 0, means not care any version update
         */
        return pushContext.pushedVersion < version && pushContext.emptyVersion == 0;
    }

    public synchronized boolean checkAndUpdatePushContext(Map<String, Long> dataCenterToPushVersionMap, Map<String, Integer> dataCenterToNumMap) {
        StringUtil.checkEquals(dataCenterToPushVersionMap.keySet(), dataCenterToNumMap.keySet(), "checkAndUpdateCtx.datacenters");
        for (Entry<String, Long> entry : dataCenterToPushVersionMap.entrySet()) {
            checkAndUpdatePushContext(entry.getKey(), entry.getValue(), dataCenterToNumMap.get(entry.getKey()));
        }
        return true;
    }

    private synchronized boolean checkAndUpdatePushContext(String dataCenter, long pushVersion, int num) {
        final PushContext pushContext = getPushContext(dataCenter);

        if (pushContext.pushedVersion < pushVersion) {
            pushContext.pushedVersion = pushVersion;
            pushContext.pushedNum = num;
            pushedFailCount = 0;
            lastPushedFailTimeStamp = 0;
            return true;
        }
        return false;
    }

    public synchronized boolean onPushFail(Map<String, Long> versions) {
        for (Entry<String, Long> entry : versions.entrySet()) {
            if (onPushFail(entry.getKey(), entry.getValue())) {
                return true;
            }
        }
        return false;
    }

    private synchronized boolean onPushFail(String dataCenter, long pushVersion) {
        final PushContext pushContext = getPushContext(dataCenter);

        if (pushContext.pushedVersion < pushVersion) {
            pushedFailCount += 1;
            lastPushedFailTimeStamp = System.currentTimeMillis();
            return true;
        }
        return false;
    }

    /**
     * return true if all datacenter check skip
     */
    public synchronized boolean checkSkipPushEmpty(Map<String, Long> versions, Map<String, Integer> nums) {
        StringUtil.checkEquals(versions.keySet(), nums.keySet(), "checkSkipPushEmpty.datacenters");

        for (Entry<String, Long> entry : versions.entrySet()) {
            if (!checkSkipPushEmpty(entry.getKey(), entry.getValue(), nums.get(entry.getKey()))) {
                return false;
            }
        }
        return true;
    }

    public synchronized boolean checkSkipPushEmpty(String dataCenter, long pushVersion, int num) {
        final PushContext pushContext = getPushContext(dataCenter);
        long lastPushMaxVersion = pushContext.lastMaxPushVersion;
        long lastPushVersion = pushContext.lastPushVersion;
        pushContext.lastMaxPushVersion = Math.max(lastPushMaxVersion, pushVersion);
        pushContext.lastPushVersion = pushVersion;
        if (num > 0) {
            return false;
        }
        if (lastPushMaxVersion <= ValueConstants.DEFAULT_NO_DATUM_VERSION) {
            return false;
        }
        return lastPushVersion == lastPushMaxVersion && lastPushMaxVersion == pushContext.pushedVersion && pushContext.pushedNum == 0;
    }

    /**
     * return true if one of any dataCenter need push empty
     */
    public synchronized boolean needPushEmpty(Set<String> dataCenters) {
        for (String dataCenter : dataCenters) {
            if (needPushEmpty(dataCenter)) {
                return true;
            }
        }
        return false;
    }

    public synchronized boolean needPushEmpty(String dataCenter) {
        final PushContext pushContext = getPushContext(dataCenter);
        if (pushContext.emptyVersion == 0) {
            return false;
        }
        // empty has mark, last push.num is not empty
        if (pushContext.pushedNum != 0) {
            return true;
        }
        // emptyVersion has pushed
        return pushContext.emptyVersion != pushContext.pushedVersion;
    }

    public synchronized boolean isMarkedPushEmpty(String dataCenter) {
        final PushContext pushContext = getPushContext(dataCenter);
        return pushContext.emptyVersion != 0;
    }

    public synchronized boolean hasPushed() {
        if (CollectionUtils.isEmpty(dataCenterToPushContextMap)) {
            return false;
        }
        for (PushContext ctx : dataCenterToPushContextMap.values()) {
            if (ctx.pushedVersion != 0) {
                return true;
            }
        }
        return false;
    }

    public String shortDesc() {
        final StringBuilder sb = new StringBuilder(256);
        sb.append("dataInfoId=").append(getDataInfoId()).append(", ");
        sb.append("registerId=").append(getRegisterId()).append(", ");
        sb.append("scope=").append(getScope()).append(", ");
        sb.append("sourceAddress=").append(getClientUrl().buildAddressString()).append(", ");
        sb.append("acceptEncodes=");
        for (String encode : getAcceptEncodes()) {
            sb.append(encode).append(",");
        }
        return sb.toString();
    }

    @Override
    @JsonIgnore
    public DataType getDataType() {
        return DataType.subscriber;
    }

    public synchronized String printPushContext() {
        return dataCenterToPushContextMap == null ? "{}" : dataCenterToPushContextMap.toString();
    }

    public synchronized long getPushedVersion(String dataCenter) {
        final PushContext ctx = getPushContext(dataCenter);
        return ctx.pushedVersion;
    }

    public synchronized int getPushedNum(String dataCenter) {
        final PushContext pushContext = getPushContext(dataCenter);
        return pushContext.pushedNum;
    }

    public synchronized long markPushEmpty(String dataCenter, long emptyVersion) {
        final PushContext pushContext = getPushContext(dataCenter);

        if (pushContext.emptyVersion != 0) {
            return pushContext.emptyVersion;
        }
        pushContext.emptyVersion = emptyVersion;
        return emptyVersion;
    }


    @JsonIgnore
    public synchronized CircuitBreakerStatistic getStatistic() {
        return new CircuitBreakerStatistic(getGroup(), getClientUrl().getIp(), getClientUrl().buildAddressString(), pushedFailCount, lastPushedFailTimeStamp);
    }

    protected Map<String, String> internAttributes(Map<String, String> attributes) {
        Map<String, String> map = super.internAttributes(attributes);
        return Maps.trimMap(map);
    }

    public String[] getAcceptEncodes() {
        if (acceptEncodes == null) {
            return ArrayUtils.EMPTY_STRING_ARRAY;
        }
        return acceptEncodes;
    }

    public void internAcceptEncoding(String acceptEncoding) {
        if (StringUtils.isBlank(acceptEncoding)) {
            this.acceptEncodes = ArrayUtils.EMPTY_STRING_ARRAY;
            return;
        }
        String[] encodes = StringUtils.split(acceptEncoding, ',');
        for (int i = 0; i < encodes.length; i++) {
            encodes[i] = StringInterner.intern(encodes[i]);
        }
        this.acceptEncodes = encodes;
    }

    @Setter
    @Getter
    private static class PushContext {
        long pushedVersion;
        long lastMaxPushVersion = -1;
        long lastPushVersion = -1;
        long emptyVersion;
        int pushedNum = -1;
    }
}
